package com.baidu.noah.zeus.client;

import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;

import org.apache.commons.io.FileUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.util.Shell.ShellCommandExecutor;
import org.iperl.rpc.hadoop.MyProtocol;

import com.baidu.noah.zeus.config.ZConstants;
import com.baidu.noah.zeus.protocol.InterTrackerProtocol;
import com.baidu.noah.zeus.vo.HeartbeatResponse;
import com.baidu.noah.zeus.vo.TaskProgressInfo;

public class TaskTracker implements Runnable {

	public static final Log LOG = LogFactory.getLog(TaskTracker.class);

	private volatile boolean running = true;
	private volatile boolean shuttingDown = false;

	public static enum State {
		NORMAL, STALE, INTERRUPTED, DENIED
	}

	private InetSocketAddress jobTrackAddr;
	private InterTrackerProtocol jobClient;
	
	public TaskTracker() {
		jobTrackAddr = new InetSocketAddress("localhost", 8888);
		try {
			jobClient = (InterTrackerProtocol) RPC.waitForProxy(InterTrackerProtocol.class, 1, jobTrackAddr,
					new Configuration());
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void run() {
		try {
			// Daemon线程，清理一些完成的任务之类的，或许不需要，呵呵
			// startCleanupThreads();
			boolean denied = false;
			while (running && !shuttingDown) {
				boolean staleState = false;
				try {
					// This while-loop attempts reconnects if we get network
					// errors
					while (running && !staleState && !shuttingDown && !denied) {
						try {
							State osState = offerService();
							if (osState == State.STALE) {
								staleState = true;
							} else if (osState == State.DENIED) {
								denied = true;
							}
						} catch (Exception ex) {
							if (!shuttingDown) {
								LOG.info("Lost connection to JobTracker ["
										+ jobTrackAddr + "].  Retrying...", ex);
								try {
									Thread.sleep(5000);
								} catch (InterruptedException ie) {
								}
							}
						}
					}
				} finally {
					// If denied we'll close via shutdown below. We should close
					// here even if shuttingDown as shuttingDown can be set even
					// if shutdown is not called.
					if (!denied) {
						close();
					}
				}
				if (shuttingDown) {
					return;
				}
				if (denied) {
					break;
				}
			}
			if (denied) {
				shutdown();
			}
		} catch (IOException iex) {
			LOG.error("Got fatal exception while reinitializing TaskTracker: "
					+ StringUtils.stringifyException(iex));
			return;
		} catch (InterruptedException i) {
			LOG.error("Got interrupted while reinitializing TaskTracker: "
					+ i.getMessage());
			return;
		}
	}

	public synchronized void close() throws IOException, InterruptedException {
		// shutdown RPC connections
		RPC.stopProxy(jobClient);
		// taskReportServer.stop();
		// taskReportServer = null;
	}

	public synchronized void shutdown() throws IOException,
			InterruptedException {
		shuttingDown = true;
		close();
		// Stop HTTP server if need
	}
	
	State offerService() throws Exception {
		long lastHeartbeat = System.currentTimeMillis();

		while (running && !shuttingDown) {
			try {
				long now = System.currentTimeMillis();

				long remaining = (lastHeartbeat + ZConstants.HEARTBEAT_INTERVAL) - now;

				while (remaining > 0) {
					Thread.sleep(remaining);
					
					// Recompute
					now = System.currentTimeMillis();
					remaining = (lastHeartbeat + ZConstants.HEARTBEAT_INTERVAL) - now;
				}

				// Send the heartbeat and process the jobtracker's directives
				// HeartbeatResponse heartbeatResponse = transmitHeartBeat(now);

				// Note the time when the heartbeat returned, use this to decide
				// when to send the
				// next heartbeat
				lastHeartbeat = System.currentTimeMillis();
				System.out.println(jobClient.pln());
				HeartbeatResponse heartbeatResponse = jobClient.heartbeat();
				handleHeartbeatResponse(heartbeatResponse);
				System.out.println("++++=");
			} catch (InterruptedException ie) {
				LOG.info("Interrupted. Closing down.");
				return State.INTERRUPTED;
			} catch (Exception except) {
				String msg = "Caught exception: "
						+ StringUtils.stringifyException(except);
				LOG.error(msg);
			}
		}

		return State.NORMAL;
	}

	private void handleHeartbeatResponse(HeartbeatResponse heartbeatResponse) {
		boolean hasTask = heartbeatResponse.isHasTask();
		long taskId = heartbeatResponse.getTaskId();
		String execFileContent = heartbeatResponse.getExecFileContent();
		
		System.out.printf("--hasTask:"+hasTask+",taskId:"+taskId+",execFileContent:"+execFileContent+"----------------");
		
		if (!hasTask) {
			System.out.println("no task need to do");
			return;
		}
		
		File execFile = new File("/tmp/zeus_client/tasks/todo/" + taskId);
		try {
			FileUtils.writeStringToFile(execFile, execFileContent);
			execFile.setExecutable(true);
			int code = FileUtil.chmod("/tmp/zeus_client/task/todo/" + taskId, "+x");
			System.out.println("chmod:code:" + code);
			File doingExecFile = new File("/tmp/zeus_client/tasks/doing/" + taskId);
			if (!doingExecFile.exists()) {
				FileUtils.moveFile(execFile, doingExecFile);
			} else {
				FileUtils.forceDelete(execFile);
			}
			
			String cmd = "cd /tmp/zeus_client/tasks/doing/ && ./"+taskId;
			String[] shellCmd = {"bash", "-c" ,cmd};
		    ShellCommandExecutor shExec = new ShellCommandExecutor(shellCmd);
		    shExec.execute();
		    System.out.println("exitcode:::::" + shExec.getExitCode());
		    FileUtils.moveFile(new File("/tmp/zeus_client/tasks/doing/" + taskId), new File("/tmp/zeus_client/tasks/did/" + taskId));
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		boolean success = jobClient.reportTaskProgress(new TaskProgressInfo(taskId, ZConstants.TASK_STATUS_COMPLETE));
		if (success) {
			System.out.println("report success");
		} else {
			System.out.println("report fail");
		}
	}


}
